package felix.things.router.mqtt;

import felix.things.router.mqtt.credentials.MqttClientCredentials;
import felix.things.router.mqtt.persistent.MqttPersistentMessage;
import felix.things.router.mqtt.persistent.PersistentFileService;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.internal.security.SSLSocketFactoryFactory;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Felix
 * @date 2019/7/27
 */
@Slf4j
@Component
public class MqttBrokerMonitor implements MqttCallback {
    private static final String DEFAULT_HOST = "localhost";
    private static final int DEFAULT_PORT = 1883;
    private static final int DEFAULT_INFLIGHT = 10;
    private static final long DEFAULT_RETRY_INTERVAL = 2000;
    private static final long DEFAULT_CONNECTION_TIMEOUT = 2000;
    private static final long DEFAULT_RATE_LIMITING = 0;

    private final UUID clientId = UUID.randomUUID();
    @Getter
    private MqttAsyncClient client;
    private MqttConnectOptions clientOptions;
    private Object connectLock = new Object();

    @Autowired
    private MqttBrokerConfiguration configuration;
    @Autowired
    private MqttClientCredentials credentials;
    @Autowired
    private PersistentFileService persistentFileService;

    @Value("${mqttrouter.enable:false}")
    private boolean enable;

    @PostConstruct
    private void init() {
        if (enable) {
            checkConfig();
            initClient();
        }
    }

    private void initClient() {
        try {
            client = new MqttAsyncClient((configuration.isSsl() ? "ssl" : "tcp") + "://" + configuration.getHost() + ":" + configuration.getPort(),
                    getClientId(), new MemoryPersistence());
            client.setCallback(this);
            clientOptions = new MqttConnectOptions();
            clientOptions.setCleanSession(true);
            clientOptions.setMaxInflight(configuration.getMaxInflight());
            if (configuration.isSsl()) {
                Properties sslProperties = new Properties();
                if (!StringUtils.isEmpty(configuration.getTruststore())) {
                    sslProperties.put(SSLSocketFactoryFactory.TRUSTSTORE, configuration.getTruststore());
                    sslProperties.put(SSLSocketFactoryFactory.TRUSTSTOREPWD, configuration.getTruststorePassword());
                    sslProperties.put(SSLSocketFactoryFactory.TRUSTSTORETYPE, "JKS");
                    sslProperties.put(SSLSocketFactoryFactory.CLIENTAUTH, false);
                }
                clientOptions.setSSLProperties(sslProperties);
            }
            credentials.configure(clientOptions);
        } catch (MqttException e) {
            log.error("[{}:{}] MQTT broker init failed!", configuration.getHost(), configuration.getPort(), e);
            throw new RuntimeException("MQTT broker init failed!", e);
        }
    }

    private String getClientId() {
        return StringUtils.isEmpty(configuration.getClientId()) ? clientId.toString() : configuration.getClientId();
    }

    public void disconnect() {
        try {
            client.disconnect();
            log.info("[{}:{}] MQTT client closed!", configuration.getHost(), configuration.getPort());
        } catch (MqttException e) {
            log.warn("[{}:{}] MQTT client disconnection faied! {}", configuration.getHost(), configuration.getPort(), e.getMessage(), e);
        }
    }

    private void checkConfig() {
        if (StringUtils.isEmpty(configuration.getHost())) {
            configuration.setHost(DEFAULT_HOST);
        }
        if (configuration.getPort() == 0) {
            configuration.setPort(DEFAULT_PORT);
        }
        if (configuration.getMaxInflight() == 0) {
            configuration.setMaxInflight(DEFAULT_INFLIGHT);
        }
        if (configuration.getRetryInterval() == 0) {
            configuration.setRetryInterval(DEFAULT_RETRY_INTERVAL);
        }
        if (configuration.getConnectionTimeout() == 0) {
            configuration.setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT);
        }
        if (configuration.getRateLimiting() == 0) {
            configuration.setConnectionTimeout(DEFAULT_RATE_LIMITING);
        }
    }

    public void checkConnection() {
        if (!client.isConnected()) {
            synchronized (connectLock) {
                while (!client.isConnected()) {
                    log.debug("[{}:{}] MQTT broker connection attempt!", configuration.getHost(), configuration.getPort());
                    try {
                        client.connect(clientOptions, null, new IMqttActionListener() {
                            @Override
                            public void onSuccess(IMqttToken iMqttToken) {
                                log.info("[{}:{}] MQTT broker connection established!", configuration.getHost(), configuration.getPort());
                                try {
                                    subscribeToTopics();
                                } catch (MqttException e) {
                                    log.error("Failed to subscribeToTopics!", e);
                                }
                            }

                            @Override
                            public void onFailure(IMqttToken iMqttToken, Throwable e) {
                                log.trace("[{}:{}] MQTT broker connection faied! {}", configuration.getHost(), configuration.getPort(), e.getMessage());
                            }
                        }).waitForCompletion(configuration.getConnectionTimeout());
                    } catch (MqttException e) {
                        log.warn("[{}:{}] MQTT broker connection faied! {}", configuration.getHost(), configuration.getPort(), e.getMessage());
                        if (!client.isConnected()) {
                            try {
                                Thread.sleep(configuration.getRetryInterval());
                            } catch (InterruptedException e1) {
                                log.error("Failed to wait for retry interval!", e);
                            }
                        }
                    }
                }
            }

        }
    }

    private Map<String, IMqttMessageListener> reconListener = new ConcurrentHashMap<>();
    private Map<String, IMqttMessageListener> connectListener = new ConcurrentHashMap<>();

    private void subscribeToTopics() throws MqttException {
        List<IMqttToken> tokens = new ArrayList<>();
        if (reconListener.size() > 0) {
            Iterator<String> it = reconListener.keySet().iterator();
            while (it.hasNext()) {
                String next = it.next();
                client.unsubscribe(next);
                tokens.add(client.subscribe(next, 1, reconListener.get(next)));
                log.info("reconnect topic 【{}】", next);
                it.remove();
            }
            for (IMqttToken token : tokens) {
                token.waitForCompletion();
            }
        }
    }

    void subscribe(String topic, IMqttMessageListener listener) {
        try {
            checkConnection();
            client.unsubscribe(topic);
            client.subscribe(topic, 1, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken iMqttToken) {
                    log.info("Successfully subscribe to topic 【{}】", topic);
                    connectListener.put(topic, listener);
                }

                @Override
                public void onFailure(IMqttToken iMqttToken, Throwable e) {
                    log.warn("Failed to subscribe to topic 【{}】", topic, e);
                    reconListener.put(topic, listener);
                }
            }, listener).waitForCompletion();
        } catch (MqttException e) {
            e.printStackTrace();
            reconListener.put(topic, listener);
        }
    }

    public void unsubscribe(String topic) {
        try {
            connectListener.remove(topic);
            checkConnection();
            client.unsubscribe(topic).waitForCompletion();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Deprecated
    public void publish(String topic, String msg) {
        publish(topic, new MqttMessage(msg.getBytes(StandardCharsets.UTF_8)));
    }

    @Deprecated
    public void publish(String topic, byte[] msg) {
        publish(topic, new MqttMessage(msg));
    }

    @Deprecated
    private void publish(String topic, MqttMessage msg) {
        try {
            client.publish(topic, msg, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken iMqttToken) {
                    log.debug("Successfully published to topic 【{}】 : data 【{}】", topic, new String(msg.getPayload()));
                }

                @Override
                public void onFailure(IMqttToken iMqttToken, Throwable e) {
                    log.warn("Failed to publish to topic 【{}】", topic, e);
                    saveForResend(topic, msg);
                }
            });
        } catch (MqttException e) {
            log.warn("Failed to publish to topic 【{}】 error 【{}】", topic, e.getMessage());
            saveForResend(topic, msg);
        }
    }

    public void publish(MqttPersistentMessage msg) {
        try {
            client.publish(msg.getTopic(), msg.getPayload(), 1, true, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken iMqttToken) {
                    log.debug("Successfully published to topic 【{}】 : data 【{}】", msg.getTopic(), new String(msg.getPayload()));
                    if (configuration.getRateLimiting() != 0) {
                        try {
                            Thread.sleep(configuration.getRateLimiting());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }

                @Override
                public void onFailure(IMqttToken iMqttToken, Throwable e) {
                    log.warn("Failed to publish to topic 【{}】", msg.getTopic(), e);
                    saveForResend(msg);
                }
            });
        } catch (MqttException e) {
            log.warn("Failed to publish to topic 【{}】 error 【{}】", msg.getTopic(), e.getMessage());
            saveForResend(msg);
        }
    }

    @Deprecated
    private void saveForResend(String topic, MqttMessage msg) {
        MqttPersistentMessage message = MqttPersistentMessage.builder().id(UUID.randomUUID())
                .timestamp(System.currentTimeMillis())
                .topic(topic).payload(msg.getPayload()).build();
        try {
            persistentFileService.saveForResend(message);
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    private void saveForResend(MqttPersistentMessage msg) {
        try {
            persistentFileService.saveForResend(msg);
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.warn("[{}:{}] MQTT broker connection lost!", configuration.getHost(), configuration.getPort());
        reconListener.clear();
        reconListener.putAll(connectListener);
        checkConnection();
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println(message);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
    }
}
